package com.atguigu.userprofile;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.constant.ConstCodes;
import com.atguigu.userprofile.dao.TagInfoDAO;
import com.atguigu.userprofile.util.MyClickhouseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class TaskBitmapApp {
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setAppName("TaskBitmapApp");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //获取业务日期
        String busiDate = args[1];

        //1.获取tagInfo表数据
        List<TagInfo> tagInfoList = TagInfoDAO.getTagInfoListWithOn();

        //2.根据tag_value_type判断类型
        //创建存放四种不同类型的List集合
        ArrayList<TagInfo> tagInfoLongList = new ArrayList<>();
        ArrayList<TagInfo> tagInfoDecimalList = new ArrayList<>();
        ArrayList<TagInfo> tagInfoStringList = new ArrayList<>();
        ArrayList<TagInfo> tagInfoDateList = new ArrayList<>();

        //3.将相同类型的tag_code放到一起，做相同的处理
        for (TagInfo tagInfo : tagInfoList) {
            if (ConstCodes.TAG_VALUE_TYPE_LONG.equals(tagInfo.getTagValueType())){
                tagInfoLongList.add(tagInfo);
            }else if (ConstCodes.TAG_VALUE_TYPE_DECIMAL.equals(tagInfo.getTagValueType())){
                tagInfoDecimalList.add(tagInfo);
            }else if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagInfo.getTagValueType())){
                tagInfoStringList.add(tagInfo);
            }else if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagInfo.getTagValueType())){
                tagInfoDateList.add(tagInfo);
            }
        }


        //4.根据类型将数据查询出来插入到Clickhouse中对应不同类型的表中
        saveBitMapToClickhouse(tagInfoLongList, "user_tag_value_long", busiDate);
        saveBitMapToClickhouse(tagInfoDecimalList, "user_tag_value_decimal", busiDate);
        saveBitMapToClickhouse(tagInfoStringList, "user_tag_value_string", busiDate);
        saveBitMapToClickhouse(tagInfoDateList, "user_tag_value_date", busiDate);
    }


    /**
     * insert into 表名
     * select
     * groupBitmapState(uid) uids,
     * tg.1,
     * tg.2
     * from (
     * select
     * uid,
     * arrayJoin([('gender',gender),('agegroup',agegroup),('favor',favor)]) as tg
     * from up_tag_merge_20200614
     * ) as utg group by tg.1,tg.2
     */
    public static void saveBitMapToClickhouse(List<TagInfo> list,String tableName,String dt){

        if (!list.isEmpty()&&list.size()>0) {

            //库名？

            List<String> tagCodeList = list.stream().map(tagInfo -> "('" + tagInfo.getTagCode().toLowerCase() + "'," + tagInfo.getTagCode().toLowerCase() + ")").collect(Collectors.toList());

            String tagCodeSQL = StringUtils.join(tagCodeList, ",");

            String insertTableName = "up_tag_merge_"+dt.replace("-", "");
            String alterSQL = "alter table " + tableName + " delete where dt ='" + dt + "'";

            MyClickhouseUtil.executeSql(alterSQL);

            System.out.println(alterSQL);

            String sql = " insert into " + tableName + "\n" +
                    "     select\n" +
                    "     tg.1,\n" +
                    "     tg.2,\n" +
                    "     groupBitmapState(cast(uid as UInt64)) us,\n" +
                    "     '" + dt + "'" +
                    "     from (\n" +
                    "    select\n" +
                    "     uid,\n" +
                    "     arrayJoin([" + tagCodeSQL + "]) as tg\n" +
                    "      from " + insertTableName + "\n" +
                    "      ) as utg group by tg.1,tg.2";

            System.out.println(sql);
            MyClickhouseUtil.executeSql(sql);
        }
    }
}
